package writer

import (
	"context"
	"github.com/segmentio/kafka-go"
	"go.uber.org/zap"
	"log"
	"v8s-ssp-proxy/internal/configuration"
	"v8s/logger"
)

type KafkaWriter struct {
	log    *zap.SugaredLogger
	writer *kafka.Writer
}

func NewKafkaWriter(properties configuration.KafkaProperties) *KafkaWriter {
	// make a writer that produces to topic-A, using the least-bytes distribution
	w := &kafka.Writer{
		Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
		Topic:    properties.ResponseTopic,
		Balancer: &kafka.LeastBytes{},
	}
	return &KafkaWriter{
		log:    logger.Logger().Named("writer").Sugar(),
		writer: w,
	}
}

func (w *KafkaWriter) KafkaWriteMsg(key, header, value string, msg []byte) {
	err := w.writer.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte(key),
			Value: []byte(value),
		},
		kafka.Message{
			Key:   []byte("Key-B"),
			Value: []byte("One!"),
		},
		kafka.Message{
			Key:   []byte("Key-C"),
			Value: []byte("Two!"),
		},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

}

func (w *KafkaWriter) Shutdown() error {
	return w.writer.Close()
}
